#!/bin/bash
# 实时监控Kinesis流中的数据

set -e

STREAM_NAME="click-events-stream"

echo "🔍 开始监控Kinesis流: $STREAM_NAME"
echo "按 Ctrl+C 停止监控"
echo "================================"

# 获取最新的Shard Iterator
SHARD_ITERATOR=$(docker exec browse-etl-localstack awslocal kinesis get-shard-iterator \
  --stream-name $STREAM_NAME \
  --shard-id shardId-000000000000 \
  --shard-iterator-type LATEST \
  --query 'ShardIterator' \
  --output text)

# 循环读取数据
while true; do
  # 获取记录
  RESULT=$(docker exec browse-etl-localstack awslocal kinesis get-records --shard-iterator $SHARD_ITERATOR)
  
  # 提取记录
  RECORDS=$(echo "$RESULT" | jq -r '.Records[]')
  
  if [ ! -z "$RECORDS" ]; then
    echo "$RESULT" | jq -r '.Records[] | "\n[\(.ApproximateArrivalTimestamp | strftime("%Y-%m-%d %H:%M:%S"))] 收到新记录:\n" + (.Data | @base64d | fromjson | tostring)'
  fi
  
  # 获取下一个iterator
  SHARD_ITERATOR=$(echo "$RESULT" | jq -r '.NextShardIterator')
  
  # 等待1秒
  sleep 1
done

